package com.pudaocredit.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;

public class HbaseConUtil {

    public static Connection getConnection(String zkQuorum, String clientPort, String keyTabPath, String krbConfPath, String principal) throws IOException {

        // krb5.conf必需
        System.setProperty("java.security.krb5.conf", krbConfPath);
        System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");

        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();

        // 必需
        conf.set("hadoop.security.authentication", "kerberos");

        // 必需
        conf.set("hbase.security.authentication", "kerberos");
        conf.set("kerberos.principal" , "hbase/ocean@PUDAO.COM" );
        conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@PUDAO.COM");
        conf.set("hbase.zookeeper.quorum", zkQuorum);
        conf.set("hbase.zookeeper.property.clientPort", clientPort);
        // (非必需)
        conf.set("hbase.master.kerberos.principal", "hbase/_HOST@PUDAO.COM");

        UserGroupInformation.setConfiguration(conf);

        //登陆认证
        UserGroupInformation.loginUserFromKeytab(principal, keyTabPath);
        UserGroupInformation loginUser = UserGroupInformation.getLoginUser();
        System.out.println("登录用户为："+loginUser);

        return ConnectionFactory.createConnection(conf);
    }

    public static void main(String[] args) throws IOException {
        Connection connection = getConnection("11.243.32.178,11.243.32.179,11.243.32.177", "2181", "/export/kerberos/1/hbase.keytab", "/etc/krb5.conf", "hbase/ocean@PUDAO.COM");
        Admin admin = connection.getAdmin();
        ClusterStatus clusterStatus = admin.getClusterStatus();
        System.out.println("Hbase版本为："+clusterStatus.getHBaseVersion());
        if (admin.tableExists(TableName.valueOf("default:phone_sha256"))) {
            System.out.println("表tonseal:tonseal_table存在");
        } else {
            System.err.println("表tonseal:tonseal_table不存在");
        }
        Table table = connection.getTable(TableName.valueOf("default:phone_sha256"));
        byte[] row = new byte[]{};
        row = "0bf366ce50c157a2750cd893c46f0d489a4efc729a6c060ce4a416e879bfa41c".getBytes();
        Get get = new Get(row);
        Result result = table.get(get);
        String valueStr = new String(result.getValue(Bytes.toBytes("phone_cf"),Bytes.toBytes("t")),"utf-8");
        System.out.println("查询hbase结果为："+valueStr);
        admin.close();
        connection.close();
    }
}